package com.example.springbootkafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.Optional;

/**
 * @ClassName LogConsumer
 * @Description TODO
 * @Auth ying.xiao
 * @Date 2021/7/13
 * @Version 1.0.0
 */
@Component
public class LogConsumer {

    @KafkaListener(topics = "log_topic")
    public void onMessage(ConsumerRecord<Integer, String> record) throws UnsupportedEncodingException {
        Optional<ConsumerRecord<Integer, String>> optional =
                Optional.ofNullable(record);
        if (optional.isPresent()) {
        System.out.println(
                    record.topic() + "\t"
                            + record.partition() + "\t"
                            + record.offset() + "\t"
                            + record.key() + "\t"
                            + URLDecoder.decode(record.value(),"utf-8"));
    }
}

}
